package com.leon.datalink.core.persistence.plugin;

import akka.persistence.journal.japi.AsyncWriteJournal;
import akka.persistence.AtomicWrite;
import akka.persistence.PersistentRepr;
import com.leon.datalink.core.evn.EnvUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.collection.immutable.Seq;
import scala.concurrent.Future;
import akka.dispatch.Futures;

import java.io.*;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;

/**
 * 真正的基于文件系统的Journal插件
 * 使用文件系统进行真实的数据持久化
 *
 * @author DataLink Team
 * @version 2.0.0
 */
public class LmdbJournalPlugin extends AsyncWriteJournal {

    private static final Logger logger = LoggerFactory.getLogger(LmdbJournalPlugin.class);

    // 文件系统存储路径
    private String persistenceDir;
    private String journalDir;

    // 内存索引（用于快速查找，但数据存储在文件中）
    private final Map<String, Long> sequenceIndex = new ConcurrentHashMap<>();
    
    public LmdbJournalPlugin() {
        // Akka会自动调用这个构造函数
    }
    
    @Override
    public void preStart() throws Exception {
        super.preStart();

        // 初始化存储路径
        persistenceDir = System.getProperty("datalink.persistence.storage.path",
                                         EnvUtil.getDatalinkHome() + "/data/persistence");
        journalDir = persistenceDir + "/journal";

        // 创建存储目录
        Files.createDirectories(Paths.get(journalDir));

        // 加载现有的序列号索引
        loadSequenceIndex();

        logger.info("Real LMDB Journal Plugin initialized with storage path: {}", journalDir);
    }
    
    @Override
    public Future<Iterable<Optional<Exception>>> doAsyncWriteMessages(Iterable<AtomicWrite> messages) {
        try {
            java.util.List<Optional<Exception>> results = new java.util.ArrayList<>();

            for (AtomicWrite atomicWrite : messages) {
                try {
                    writeAtomicWriteToFile(atomicWrite);
                    results.add(Optional.empty()); // 成功
                } catch (Exception e) {
                    logger.error("Failed to write atomic write: {}", atomicWrite, e);
                    results.add(Optional.of(e)); // 失败
                }
            }

            return Futures.successful((Iterable<Optional<Exception>>) results);
        } catch (Exception e) {
            logger.error("Failed to write messages", e);
            return Futures.failed(e);
        }
    }
    
    @Override
    public Future<Void> doAsyncDeleteMessagesTo(String persistenceId, long toSequenceNr) {
        try {
            // 删除指定序列号之前的消息文件
            for (long seq = 1; seq <= toSequenceNr; seq++) {
                String filePath = getJournalFilePath(persistenceId, seq);
                File file = new File(filePath);
                if (file.exists()) {
                    boolean deleted = file.delete();
                    if (!deleted) {
                        logger.warn("Failed to delete journal file: {}", filePath);
                    }
                }

                // 从索引中移除
                String key = createJournalKey(persistenceId, seq);
                sequenceIndex.remove(key);
            }

            logger.debug("Deleted messages for persistenceId: {} up to sequence: {}",
                       persistenceId, toSequenceNr);
            return Futures.successful(null);
        } catch (Exception e) {
            logger.error("Failed to delete messages for persistenceId: {}", persistenceId, e);
            return Futures.failed(e);
        }
    }
    
    @Override
    public Future<Void> doAsyncReplayMessages(String persistenceId, long fromSequenceNr,
                                            long toSequenceNr, long max,
                                            Consumer<PersistentRepr> recoveryCallback) {
        try {
            long count = 0;
            for (long seq = fromSequenceNr; seq <= toSequenceNr && count < max; seq++) {
                PersistentRepr repr = readPersistentReprFromFile(persistenceId, seq);
                if (repr != null) {
                    recoveryCallback.accept(repr);
                    count++;
                }
            }
            logger.debug("Replayed {} messages for persistenceId: {} from {} to {}",
                       count, persistenceId, fromSequenceNr, toSequenceNr);
            return Futures.successful(null);
        } catch (Exception e) {
            logger.error("Failed to replay messages for persistenceId: {}", persistenceId, e);
            return Futures.failed(e);
        }
    }
    
    @Override
    public Future<Long> doAsyncReadHighestSequenceNr(String persistenceId, long fromSequenceNr) {
        try {
            // 从索引中查找最高序列号
            long highestSeq = fromSequenceNr - 1;

            for (Map.Entry<String, Long> entry : sequenceIndex.entrySet()) {
                String key = entry.getKey();
                if (key.startsWith(persistenceId + "-")) {
                    long seq = entry.getValue();
                    if (seq >= fromSequenceNr && seq > highestSeq) {
                        highestSeq = seq;
                    }
                }
            }

            logger.debug("Highest sequence number for persistenceId: {} is {}",
                       persistenceId, highestSeq);
            return Futures.successful(highestSeq);
        } catch (Exception e) {
            logger.error("Failed to read highest sequence number for persistenceId: {}",
                       persistenceId, e);
            return Futures.successful(fromSequenceNr - 1);
        }
    }
    
    /**
     * 将AtomicWrite写入文件系统
     */
    private void writeAtomicWriteToFile(AtomicWrite atomicWrite) throws Exception {
        scala.collection.Iterator<PersistentRepr> iterator = atomicWrite.payload().iterator();

        while (iterator.hasNext()) {
            PersistentRepr repr = iterator.next();
            writePersistentReprToFile(repr);

            // 更新索引
            String key = createJournalKey(repr.persistenceId(), repr.sequenceNr());
            sequenceIndex.put(key, repr.sequenceNr());
        }

        // 保存索引到文件
        saveSequenceIndex();
    }
    
    /**
     * 将PersistentRepr写入文件
     */
    private void writePersistentReprToFile(PersistentRepr repr) throws Exception {
        String filePath = getJournalFilePath(repr.persistenceId(), repr.sequenceNr());

        // 确保目录存在
        File file = new File(filePath);
        file.getParentFile().mkdirs();

        // 序列化并写入文件
        try (ObjectOutputStream oos = new ObjectOutputStream(new FileOutputStream(file))) {
            // 写入基本信息
            oos.writeUTF(repr.persistenceId());
            oos.writeLong(repr.sequenceNr());
            oos.writeUTF(repr.manifest());
            oos.writeUTF(repr.writerUuid());

            // 写入payload（事件数据）
            if (repr.payload() instanceof Serializable) {
                oos.writeObject(repr.payload());
            } else {
                // 如果payload不可序列化，写入其字符串表示
                oos.writeObject(repr.payload().toString());
            }

            oos.flush();
        }

        logger.debug("Wrote PersistentRepr to file: {}", filePath);
    }

    /**
     * 从文件读取PersistentRepr
     */
    private PersistentRepr readPersistentReprFromFile(String persistenceId, long sequenceNr) {
        String filePath = getJournalFilePath(persistenceId, sequenceNr);
        File file = new File(filePath);

        if (!file.exists()) {
            return null;
        }

        try (ObjectInputStream ois = new ObjectInputStream(new FileInputStream(file))) {
            String readPersistenceId = ois.readUTF();
            long readSequenceNr = ois.readLong();
            String manifest = ois.readUTF();
            String writerUuid = ois.readUTF();
            Object payload = ois.readObject();

            // 重建PersistentRepr
            return PersistentRepr.apply(
                payload,
                readSequenceNr,
                readPersistenceId,
                manifest,
                false, // deleted
                null, // sender
                writerUuid
            );

        } catch (Exception e) {
            logger.error("Failed to read PersistentRepr from file: {}", filePath, e);
            return null;
        }
    }

    /**
     * 获取Journal文件路径
     */
    private String getJournalFilePath(String persistenceId, long sequenceNr) {
        // 创建分层目录结构以避免单个目录文件过多
        String subDir = persistenceId.replaceAll("[^a-zA-Z0-9]", "_");
        return journalDir + "/" + subDir + "/" + String.format("%020d", sequenceNr) + ".journal";
    }

    /**
     * 创建Journal键
     */
    private String createJournalKey(String persistenceId, long sequenceNr) {
        return persistenceId + "-" + String.format("%020d", sequenceNr);
    }

    /**
     * 加载序列号索引
     */
    private void loadSequenceIndex() {
        String indexFile = journalDir + "/sequence-index.dat";
        File file = new File(indexFile);

        if (!file.exists()) {
            logger.info("Sequence index file not found, starting with empty index");
            return;
        }

        try (ObjectInputStream ois = new ObjectInputStream(new FileInputStream(file))) {
            @SuppressWarnings("unchecked")
            Map<String, Long> loadedIndex = (Map<String, Long>) ois.readObject();
            sequenceIndex.putAll(loadedIndex);
            logger.info("Loaded sequence index with {} entries", sequenceIndex.size());
        } catch (Exception e) {
            logger.error("Failed to load sequence index", e);
        }
    }

    /**
     * 保存序列号索引
     */
    private void saveSequenceIndex() {
        String indexFile = journalDir + "/sequence-index.dat";

        try (ObjectOutputStream oos = new ObjectOutputStream(new FileOutputStream(indexFile))) {
            oos.writeObject(new HashMap<>(sequenceIndex));
            oos.flush();
        } catch (Exception e) {
            logger.error("Failed to save sequence index", e);
        }
    }

    @Override
    public void postStop() throws Exception {
        // 保存索引
        saveSequenceIndex();

        // 清理内存索引
        if (sequenceIndex != null) {
            sequenceIndex.clear();
        }

        logger.info("Real LMDB Journal Plugin stopped");
        super.postStop();
    }
}
